package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo12Join {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo12Join")

    val sc: SparkContext = new SparkContext(conf)

    // 分别加载学生以及分数的数据，并将其都变成KV格式
    // 将关联的条件作为Key
    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")
    val scoRDD: RDD[String] = sc.textFile("spark/data/stu/score.txt")

    val stuKVRDD: RDD[(String, String)] = stuRDD
      .map(line => (line.split(",")(0), line))

    val scoKVRDD: RDD[(String, String)] = scoRDD
      .map(line => (line.split(",")(0), line))

    val joinRDD: RDD[(String, (String, String))] = stuKVRDD.join(scoKVRDD)
    joinRDD
    //      .foreach(println)

    joinRDD.map(kv => {
      val id: String = kv._1
      val stuSplits: Array[String] = kv._2._1.split(",")
      val scoSplits: Array[String] = kv._2._2.split(",")
      s"$id,${stuSplits(1)},${stuSplits(4)},${scoSplits(1)},${scoSplits(2)}"
    })
    //      .foreach(println)

    val leftJoinRDD: RDD[(String, (String, Option[String]))] = stuKVRDD.leftOuterJoin(scoKVRDD)
    leftJoinRDD
      .map(kv => {
        val id: String = kv._1
        val stuSplits: Array[String] = kv._2._1.split(",")
        val scoStr: String = kv._2._2 match {
          case Some(v) =>
            val scoSplits: Array[String] = v.split(",")
            s"${scoSplits(1)},${scoSplits(2)}"
          case None =>
            "None"
        }

        s"$id,${stuSplits(1)},${stuSplits(4)},$scoStr"
      })
    //      .foreach(println)

    val rightJoinRDD: RDD[(String, (Option[String], String))] = stuKVRDD.rightOuterJoin(scoKVRDD)

    rightJoinRDD.map(kv => {
      val id: String = kv._1
      val scoSplits: Array[String] = kv._2._2.split(",")
      val stuStr: String = kv._2._1 match {
        case Some(v) =>
          val stuSplits: Array[String] = v.split(",")
          s"${stuSplits(1)},${stuSplits(4)}"
        case None =>
          "None"
      }

      s"$id,$stuStr,${scoSplits(1)},${scoSplits(2)}"
    })
    //      .foreach(println)

    stuKVRDD.fullOuterJoin(scoKVRDD)
      .map(kv => {
        val id: String = kv._1
        val stuStr: String = kv._2._1 match {
          case Some(v) =>
            val stuSplits: Array[String] = v.split(",")
            s"${stuSplits(1)},${stuSplits(4)}"
          case None =>
            "None"
        }

        val scoStr: String = kv._2._2 match {
          case Some(v) =>
            val scoSplits: Array[String] = v.split(",")
            s"${scoSplits(1)},${scoSplits(2)}"
          case None =>
            "None"
        }

        s"$id,$stuStr,$scoStr"
      })
      .foreach(println)
  }

}
